RocketMQ Connect 實戰 2
PostgreSQL 來源 (CDC) - >RocketMQ Connect -> MySQL 儲存槽 (JDBC)
準備
啟動 RocketMQ
- Linux/Unix/Mac
- 64 位元 JDK 1.8+；
- Maven 3.2.x+；
- 啟動 RocketMQ；
提示：${ROCKETMQ_HOME} 位置說明
bin-release.zip 版本：/rocketmq-all-4.9.4-bin-release
source-release.zip 版本：/rocketmq-all-4.9.4-source-release/distribution
啟動 Connect
編譯 Connector 外掛程式
Debezium RocketMQ Connector
$ cd rocketmq-connect/connectors/rocketmq-connect-debezium/
$ mvn clean package -Dmaven.test.skip=true
將編譯好的 Debezium PostgreSQL RocketMQ Connector 套件移至執行階段載入目錄。指令如下：
mkdir -p /usr/local/connector-plugins
cp rocketmq-connect-debezium-postgresql/target/rocketmq-connect-debezium-postgresql-0.0.1-SNAPSHOT-jar-with-dependencies.jar /usr/local/connector-plugins
JDBC Connector
將編譯好的 JDBC Connector 套件移至執行階段載入目錄。指令如下：
$ cd rocketmq-connect/connectors/rocketmq-connect-jdbc/
$ mvn clean package -Dmaven.test.skip=true
cp rocketmq-connect-jdbc/target/rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar /usr/local/connector-plugins
啟動 Connect Runtime
cd rocketmq-connect
mvn -Prelease-connect -DskipTests clean install -U
修改設定檔 connect-standalone.conf，主要設定如下
$ cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
$ vim conf/connect-standalone.conf
$ cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
$ vim conf/connect-standalone.conf
workerId=standalone-worker
storePathRootDir=/tmp/storeRoot
## Http port for user to access REST API
httpPort=8082
# Rocketmq namesrvAddr
namesrvAddr=localhost:9876
# RocketMQ acl
aclEnable=false
accessKey=rocketmq
secretKey=12345678
autoCreateGroupEnable=false
clusterName="DefaultCluster"
# Core configuration, configure the plugin directory of the previously compiled debezium package here
# Source or sink connector jar file dir,The default value is rocketmq-connect-sample
pluginPaths=/usr/local/connector-plugins
cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
sh bin/connect-standalone.sh -c conf/connect-standalone.conf &
Postgres image
使用 debezium 的 Postgres docker 環境建立 Postgres 資料庫
# starting a pg instance
docker run -d --name postgres -p 5432:5432 -e POSTGRES_USER=start_data_engineer -e POSTGRES_PASSWORD=password debezium/postgres:14
# bash into postgres instance
docker exec -ti postgres /bin/bash
Postgres 資訊 Port：5432 帳號：start_data_engineer/password 同步來源資料庫：bank.holding 目標資料庫表格：bank1.holding
MySQL image
使用 debezium 的 MySQL docker 環境建立 MySQL 資料庫
docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw quay.io/debezium/example-mysql:1.9
MySQL 資訊
Port：3306
帳號：root/debezium
測試資料
使用帳號 start_data_engineer/password 登入資料庫
來源資料庫表格：bank.holding
CREATE SCHEMA bank;
SET search_path TO bank,public;
CREATE TABLE bank.holding (
holding_id int,
user_id int,
holding_stock varchar(8),
holding_quantity int,
datetime_created timestamp,
datetime_updated timestamp,
primary key(holding_id)
);
ALTER TABLE bank.holding replica identity FULL;
insert into bank.holding values (1000, 1, 'VFIAX', 10, now(), now());
\q
insert into bank.holding values (1000, 1, 'VFIAX', 10, now(), now());
insert into bank.holding values (1001, 2, 'SP500', 1, now(), now());
insert into bank.holding values (1003, 3, 'SP500', 1, now(), now());
update bank.holding set holding_quantity = 300 where holding_id=1000;
目標資料庫表格：bank1.holding
create database bank1;
CREATE TABLE holding (
holding_id int,
user_id int,
holding_stock varchar(8),
holding_quantity int,
datetime_created bigint,
datetime_updated bigint,
primary key(holding_id)
);
啟動 Connector
啟動 Debezium source connector
同步來源表格資料：bank.holding 目的：解析 Postgres binlog 並封裝成共用的 ConnectRecord 物件，傳送至 RocketMQ Topic
curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/postgres-connector -d '{
"connector.class": "org.apache.rocketmq.connect.debezium.postgres.DebeziumPostgresConnector",
"max.task": "1",
"connect.topicname": "debezium-postgres-source-01",
"kafka.transforms": "Unwrap",
"kafka.transforms.Unwrap.delete.handling.mode": "none",
"kafka.transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"kafka.transforms.Unwrap.add.headers": "op,source.db,source.table",
"database.history.skip.unparseable.ddl": true,
"database.server.name": "bankserver1",
"database.port": 5432,
"database.hostname": "database ip",
"database.connectionTimeZone": "UTC",
"database.user": "start_data_engineer",
"database.dbname": "start_data_engineer",
"database.password": "password",
"table.whitelist": "bank.holding",
"key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
}'
啟動 JDBC sink connector
目的：從 Topic 消費資料並透過 JDBC 協定寫入目標表格
curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/jdbcmysqlsinktest201 -d '{
"connector.class": "org.apache.rocketmq.connect.jdbc.connector.JdbcSinkConnector",
"max.task": "2",
"connect.topicnames": "debezium-postgres-source-01",
"connection.url": "jdbc:mysql://database ip:3306/bank1",
"connection.user": "root",
"connection.password": "debezium",
"pk.fields": "holding_id",
"table.name.from.header": "true",
"pk.mode": "record_key",
"insert.mode": "UPSERT",
"db.timezone": "UTC",
"table.types": "TABLE",
"errors.deadletterqueue.topic.name": "dlq-topic",
"errors.log.enable": "true",
"errors.tolerance": "ALL",
"delete.enabled": "true",
"key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
}'
建立上述兩個 Connector 任務後，使用帳號 start_data_engineer/password 登入資料庫
對來源資料庫表格 bankholding 進行任何新增、刪除或修改，都會同步至目標表格 bank1.holding